package com.demo.component;


import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Optional;

@Component
public class KafkaConsumer {
    // 消费监听
    @KafkaListener(topics = {"topic1"})
    public void onMessage1(List<ConsumerRecord<?,?>> records,Acknowledgment ack){
        try {
            for (ConsumerRecord<?, ?> record : records) {
                Optional<?> kafkaMessage = Optional.ofNullable(record.value());
                if (kafkaMessage.isPresent()) {
                    System.out.println("简单消费："+record.topic()+"-"+record.partition()+"-"+record.value());
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            ack.acknowledge();
        }
    }
}
